# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: helpers\worker\status_manager\__init__.py
import time, threading, logging
from sqlalchemy import select as sql_select
from db import Connection
from db.tables.workers import WorkerRow, WorkersTable, WorkerAuthorization, WorkerStatus
from helpers.http_clients.scan_api import ScanApiClient
from helpers.http_clients.exceptions import *
from db.tables.events import create_event, EventResources

_logger = logging.getLogger("helpers.worker.status_manager")
time_delta_for_offline = 180


class WorkerStatusEntry:
    OFFLINE = 0
    ERROR = 1
    OK = 2

    def __init__(self):
        self._lock = threading.Lock()
        self.status_history = []
        self.notified = False

    def update(self, status, extra):
        """"
        don't maintain entries older than 10 minutes
        """
        with self._lock:
            t = time.time()
            self.status_history.append((t, status, extra))
            if status == self.OK:
                self.notified = False
            for i in range(0, len(self.status_history)):
                if self.status_history[i][0] + time_delta_for_offline + 60 > t:
                    self.status_history = self.status_history[i:]
                    break

    @property
    def last_status(self):
        with self._lock:
            if self.status_history:
                return self.status_history[-1]

    @property
    def is_offline(self):
        """
        we consider the worker offline if was not reachable in the last 10 minutes
        :return:
        """
        t = time.time()
        with self._lock:
            if not self.status_history:
                return False
            for i in range(0, len(self.status_history)):
                if self.status_history[i][1] == self.OK:
                    return False

            if self.status_history[0][0] + time_delta_for_offline > t:
                return False
            return True


class WorkerStatusStorage:
    def __init__(self):
        self.storage = dict()
        self._lock = threading.Lock()

    def update(self, worker_id, status, extra=None):
        with self._lock:
            try:
                self.storage[worker_id].update(status, extra)
            except KeyError:
                ws = WorkerStatusEntry()
                ws.update(status, extra)
                self.storage[worker_id] = ws

    def get(self, worker_id):
        with self._lock:
            return self.storage.get(worker_id)


def mark_worker_detached(worker, shard, reason):
    print("mark_worker_detached", worker, reason)
    with Connection(shard) as (db):
        q = (
            WorkersTable.update(
                values=dict(
                    authorization=WorkerAuthorization.DETACHED,
                    status=WorkerStatus.DETACHED,
                )
            )
        ).where(WorkerRow.worker_id == worker["worker_id"])
        db.execute(q)
        from db.data.users import system_users

        event = create_event(
            "worker_detached",
            system_users[0],
            system_users[0],
            resource_type=EventResources.worker,
            resource_id=worker["worker_id"],
            data=dict(
                worker_decription=worker["description"],
                endpoint=worker["endpoint"],
                reason=reason,
            ),
            shard=shard,
        )
        db.execute(event)


def notify_worker_not_responding(wss, worker, shard):
    worker_status = wss.get(worker["worker_id"])
    if worker_status is None:
        return
    if worker_status.is_offline:
        if not worker_status.notified:
            print("notify_worker_offline", worker)
            with Connection(shard) as (db):
                from db.data.users import system_users

                event = create_event(
                    "worker_not_responding",
                    system_users[0],
                    system_users[0],
                    resource_type=EventResources.worker,
                    resource_id=worker["worker_id"],
                    data=dict(
                        worker_decription=worker["description"],
                        endpoint=worker["endpoint"],
                    ),
                    shard=shard,
                )
                db.execute(event)
            worker_status.notified = True
        with Connection(shard) as (db):
            q = (WorkersTable.update(values=dict(status=WorkerStatus.OFFLINE))).where(
                WorkerRow.worker_id == worker["worker_id"]
            )
            db.execute(q)


def check_workers(wss, shard="master"):
    workers = []
    with Connection(shard) as (db):
        q = sql_select((WorkerRow,)).where(
            WorkerRow.authorization == WorkerAuthorization.AUTHORIZED
        )
        for worker in db.execute(q).fetchall():
            workers.append(dict(worker))

    for worker in workers:
        client = ScanApiClient(
            worker["endpoint"],
            worker["scanning_app"],
            api_key=worker["api_key"],
            request_timeout=5,
            ssl_verify=worker["ssl_verify"],
        )
        try:
            worker_status = client.get_worker_status()
        except AuthException:
            _logger.warning("AuthException %s", worker["worker_id"])
            wss.update(worker["worker_id"], WorkerStatusEntry.ERROR, "detached")
            mark_worker_detached(worker, shard, "error: worker authentication failed")
        except HTTP404:
            _logger.warning("HTTP404 %s", worker["worker_id"])
            wss.update(worker["worker_id"], WorkerStatusEntry.ERROR, "detached")
            mark_worker_detached(worker, shard, "error: invalid worker response")
        except HTTP500:
            _logger.warning("HTTP500 %s", worker["worker_id"])
            wss.update(
                worker["worker_id"], WorkerStatusEntry.ERROR, "worker internal error"
            )
            notify_worker_not_responding(wss, worker, shard)
        except HTTPException:
            _logger.warning("HTTPException %s", worker["worker_id"])
            wss.update(
                worker["worker_id"], WorkerStatusEntry.ERROR, "request exception"
            )
            notify_worker_not_responding(wss, worker, shard)
        except RequestError:
            _logger.warning("RequestError %s", worker["worker_id"])
            wss.update(worker["worker_id"], WorkerStatusEntry.OFFLINE, "request error")
            notify_worker_not_responding(wss, worker, shard)
        else:
            wss.update(worker["worker_id"], WorkerStatusEntry.OK)
            updates = dict()
            changes = False
            if WorkerStatus.ONLINE != worker["status"]:
                updates["status"] = WorkerStatus.ONLINE
                changes = True
            updates["info"] = worker["info"] if worker["info"] else dict()
            if worker_status.app_version != updates["info"].get("app_version"):
                updates["info"]["app_version"] = worker_status.app_version
                changes = True
            updates["info"]["license_hash"] = (
                worker_status.license_hash != updates["info"].get("license_hash")
                and worker_status
            ).license_hash
            changes = True

        if changes:
            with Connection(shard) as (db):
                q = (WorkersTable.update(values=updates)).where(
                    WorkerRow.worker_id == worker["worker_id"]
                )
                db.execute(q)
